package cn.com.sparkle.paxos.net;

import java.io.IOException;
import java.io.Serializable;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import com.google.protobuf.GeneratedMessage;

import cn.com.sparkle.paxos.AccountBook;
import cn.com.sparkle.paxos.ClusterState;
import cn.com.sparkle.paxos.ConfigNode;
import cn.com.sparkle.paxos.Configuration;
import cn.com.sparkle.paxos.NodeState;
import cn.com.sparkle.paxos.NodesCollection;
import cn.com.sparkle.paxos.addprocess.AddRequestDealer;
//import cn.com.sparkle.paxos.client.net.message.AddRequest;
//import cn.com.sparkle.paxos.client.net.message.GetNewsSenatorsRequest;
import cn.com.sparkle.paxos.event.CatchUpEvent;
import cn.com.sparkle.paxos.event.ConfigureEvent;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.CatchUpRecord;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.CatchUpRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.CatchUpResponse;
//import cn.com.sparkle.paxos.model.ElectionId;
//import cn.com.sparkle.paxos.model.Value;
//import cn.com.sparkle.paxos.net.message.GetNewsSenatorsResponse;
//import cn.com.sparkle.paxos.net.message.LookUpLatestInstanceIdRequest;
//import cn.com.sparkle.paxos.net.message.LookUpLatestInstanceIdResponse;
//import cn.com.sparkle.paxos.net.message.MessagePackage;
//import cn.com.sparkle.paxos.net.message.PrepareResponseBad;
//import cn.com.sparkle.paxos.net.message.PrepareResponseGood;
//import cn.com.sparkle.paxos.net.message.SenatorHeatBeatRequest;
//import cn.com.sparkle.paxos.net.message.SenatorHeatBeatResponse;
//import cn.com.sparkle.paxos.net.message.VoteResponse;
//import cn.com.sparkle.paxos.net.message.election.ElectionPrepareRequest;
//import cn.com.sparkle.paxos.net.message.election.ElectionSuccessMessage;
//import cn.com.sparkle.paxos.net.message.election.ElectionVoteRequest;
//import cn.com.sparkle.paxos.net.message.instance.InstancePrepareRequest;
//import cn.com.sparkle.paxos.net.message.instance.InstanceSuccessMessage;
//import cn.com.sparkle.paxos.net.message.instance.InstanceVoteRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ElectionId;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ElectionPrepareRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ElectionSuccessMessage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ElectionVoteRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.GetNewsSenatorsResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Id;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstancePrepareRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstanceSuccessMessage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstanceVoteRecord;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstanceVoteRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.LookUpLatestInstanceIdResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.PrepareResponseBad;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.PrepareResponseGood;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SenatorHeatBeatResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SuccessfulRecord;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Value;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ValueType;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.VoteResponse;
import cn.com.sparkle.paxos.stablestorage.ReadSuccessRecordCallback;
import cn.com.sparkle.paxos.stablestorage.event.PrepareRecordRealWriteEvent;
import cn.com.sparkle.paxos.util.IdComparator;
//import cn.com.sparkle.paxos.stablestorage.model.InstancePrepareRecord;
//import cn.com.sparkle.paxos.stablestorage.model.InstanceVoteRecord;
//import cn.com.sparkle.paxos.stablestorage.model.InstanceVotedRecord;
//import cn.com.sparkle.paxos.stablestorage.model.SuccessfulRecord;
import cn.com.sparkle.raptor.core.buff.IoBuffer;
import cn.com.sparkle.raptor.core.buff.SyncBuffPool;
import cn.com.sparkle.raptor.core.protocol.MultiThreadProtecolHandler.ProtocolHandlerIoSession;
import cn.com.sparkle.raptor.core.protocol.Protocol;
import cn.com.sparkle.raptor.core.protocol.ProtocolHandler;
import cn.com.sparkle.raptor.core.transport.socket.nio.IoSession;
import cn.com.sparkle.raptor.core.transport.socket.nio.exception.SessionHavaClosedException;

public class ServerNetHandler implements ProtocolHandler {
	private final static Logger logger = Logger
			.getLogger(ServerNetHandler.class);
	private final static int MAX_CATCH_SIZE = 1024 * 128;

	private volatile ClusterState cState = null;
	private volatile AccountBook aBook = null;
	private EventsManager eventsManager;

	private ReentrantLock connectLock = new ReentrantLock();
	private HashSet<ProtocolHandlerIoSession> activeSession = new HashSet<ProtocolHandlerIoSession>();
	private boolean connectAble = true;

	private final static class SenatorFollowerVersion {
		private Set<ConfigNode> senator;
		private Set<ConfigNode> follower;
		private long version;
	}

	private volatile SenatorFollowerVersion senatorVersion;

	public ServerNetHandler(EventsManager eventsManager, Configuration conf) {
		this.eventsManager = eventsManager;
		this.senatorVersion = new SenatorFollowerVersion();
		this.senatorVersion.senator = conf.getSenators();
		this.senatorVersion.version = conf.getVersion();
		this.senatorVersion.follower = conf.getFollowers();
		eventsManager.registerEvent(new ConfigureEvent() {
			@Override
			public void senatorsChange(Set<ConfigNode> newSenators, long version) {
				synchronized (this) {
					SenatorFollowerVersion newTemp = new SenatorFollowerVersion();
					newTemp.senator = newSenators;
					newTemp.follower = senatorVersion.follower;
					newTemp.version = version;
					senatorVersion = newTemp;
				}
			}

			@Override
			public void followersChange(Set<ConfigNode> newFollowers) {
				synchronized (this) {
					SenatorFollowerVersion newTemp = new SenatorFollowerVersion();
					newTemp.senator = senatorVersion.senator;
					newTemp.follower = newFollowers;
					newTemp.version = senatorVersion.version;
					senatorVersion = newTemp;
				}
			}
		});

		eventsManager.registerEvent(new CatchUpEvent() {
			@Override
			public void recoveryFromFail() {
				try {
					connectLock.lock();
					connectAble = true;
				} finally {
					connectLock.unlock();
				}
			}

			@Override
			public void catchUpFail() {
				try {
					connectLock.lock();
					logger.debug("catch up fail,and close all connection from others of senator");
					connectAble = false;
					// close all connection from others
					for (ProtocolHandlerIoSession session : activeSession) {
						session.closeSocketChannel();
					}
					activeSession.clear();
				} finally {
					connectLock.unlock();
				}
			}

		});
	}

	public void setcState(ClusterState cState) {
		this.cState = cState;
	}

	public void setaBook(AccountBook aBook) {
		this.aBook = aBook;
	}

	@Override
	public void onOneThreadSessionOpen(ProtocolHandlerIoSession session) {
		try {
			connectLock.lock();
			if (connectAble) {
				activeSession.add(session);
			} else {
				session.closeSocketChannel();
			}
		} finally {
			connectLock.unlock();
		}

	}

	@Override
	public void onOneThreadSessionClose(ProtocolHandlerIoSession session) {
		try {
			connectLock.lock();
			activeSession.remove(session);
		} finally {
			connectLock.unlock();
		}
	}

	@Override
	public void onOneThreadCatchException(IoSession session,
			ProtocolHandlerIoSession attachment, Throwable e) {
		logger.error("except close connection", e);
	}

	@Override
	public void onOneThreadMessageRecieved(Object object,
			final ProtocolHandlerIoSession session) {
		while (cState == null || aBook == null) {
			// wait system initialization be finished.
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
			}
		}

		final MessagePackage messagePackage = (MessagePackage) object;
		MessagePackage.Builder responseBuilder = null;
		if (messagePackage.hasSenatorHeartBeatRequest()) {
			ElectionId id = cState.getLastElectionId();
			NodeState masterState = cState.getSenators().getNodeStates()
					.get(id.getId().getAddress());
			boolean isMasterConnected = masterState == null ? false
					: masterState.isConnected();

			SenatorHeatBeatResponse.Builder builder = SenatorHeatBeatResponse
					.newBuilder();
			builder.setLastCanExecuteInstanceId(aBook
					.getLastCanExecutableInstanceId());
			builder.setIsMasterConnected(id.getId().getIncreaseId() < 0 ? false
					: isMasterConnected);
			builder.setElectionAddress(id.getId().getAddress());
			builder.setElectionId(id.getId().getIncreaseId());
			builder.setElectionVersion(id.getVersion());
			responseBuilder = MessagePackage.newBuilder();
			responseBuilder.setSenatorHeartBeatResponse(builder);
		} else if (messagePackage.hasElectionPrepareRequest()) {
			ElectionPrepareRequest request = messagePackage
					.getElectionPrepareRequest();
			ElectionId id = cState.getSelfState().casElectionPrepareId(
					request.getId());
			responseBuilder = MessagePackage.newBuilder();
			if (id == request.getId()) {
				Value t = Value.newBuilder().setType(ValueType.TYPE_ADMIN)
						.build();
				PrepareResponseGood response = PrepareResponseGood.newBuilder()
						.setLastVotedId(request.getId().getId()).setValue(t)
						.build();
				responseBuilder.setPrepareResponseGood(response);
				// response = new PrepareResponseGood(request.getVoteId(),
				// null);
			} else {
				PrepareResponseBad response = PrepareResponseBad.newBuilder()
						.setLastPrepareId(id.getId().getIncreaseId()).build();
				// response = new PrepareResponseBad(id.getIncreaseId());
				responseBuilder.setPrepareResponseBad(response);
			}
		} else if (messagePackage.hasElectionVoteRequest()) {
			ElectionVoteRequest request = messagePackage
					.getElectionVoteRequest();
			ElectionId id = cState.getSelfState().casElectionPrepareId(
					request.getId());
			VoteResponse response;
			if (id == request.getId()) {
				response = VoteResponse.newBuilder().setRefuseId(-1).build();
			} else {
				response = VoteResponse.newBuilder()
						.setRefuseId(id.getId().getIncreaseId()).build();
			}
			responseBuilder = MessagePackage.newBuilder();
			responseBuilder.setVoteResponse(response);
		} else if (messagePackage.hasElectionSuccessMessage()) {
			ElectionSuccessMessage message = messagePackage
					.getElectionSuccessMessage();
			ElectionId id = message.getId();
			cState.changeLastElectionId(id);
		} else if (messagePackage.hasInstanceSuccessMessage()) {
			InstanceSuccessMessage message = messagePackage
					.getInstanceSuccessMessage();
			try {
				SuccessfulRecord record = SuccessfulRecord.newBuilder()
						.setV(message.getValue()).build();
				// SuccessfulRecord record = new SuccessfulRecord(
				// message.getInstanceId(),
				// (Value) message.getValue(), null);
				aBook.writeSuccessfulRecord(message.getId(), record, null);
				logger.debug("instanceId:" + message.getId() + " succeeded");
			} catch (IOException e) {
				logger.error("fatal error", e);
			}
		} else if (messagePackage.hasLookUpLatestInstanceIdRequest()) {
			LookUpLatestInstanceIdResponse response = LookUpLatestInstanceIdResponse
					.newBuilder().setInstanceId(aBook.getMaxInstanceIdInVote())
					.build();
			responseBuilder = MessagePackage.newBuilder();
			responseBuilder.setLookUpLatestInstanceIdResponse(response);
		} else if (messagePackage.hasInstancePrepareRequest()) {

			InstancePrepareRequest request = messagePackage
					.getInstancePrepareRequest();
			logger.debug("prepare instanceId:" + request.getInstanceId());
			try {
				if (aBook.isSuccessful(request.getInstanceId())) {
					PrepareResponseBad response = PrepareResponseBad
							.newBuilder().setLastPrepareId(Long.MAX_VALUE)
							.build();
					responseBuilder = MessagePackage.newBuilder();
					responseBuilder.setPrepareResponseBad(response);
					logger.debug("prepare refused:the instance has succeeded.");
				} else {
					// InstancePrepareRecord record =
					//
					// new InstancePrepareRecord(
					// request.getInstanceId(), request.getVoteId());
					final long ct = System.currentTimeMillis();
					long result = aBook.writePrepareRecord(
							request.getInstanceId(), request.getId(),
							new PrepareRecordRealWriteEvent() {
								@Override
								public void call(long instanceId,
										InstanceVoteRecord record) {
									Id id = record.hasHighestVotedNum() ? record
											.getHighestVotedNum() : record
											.getHighestJoinNum();
									PrepareResponseGood.Builder response = PrepareResponseGood
											.newBuilder().setLastVotedId(id);
									if (record.hasHighestValue()) {
										response.setValue(record
												.getHighestValue());
									}
									MessagePackage.Builder builder = MessagePackage
											.newBuilder();
									builder.setPrepareResponseGood(response);
									sendResponse(session, builder,
											messagePackage.getId());
									logger.debug("instanceId: "
											+ instanceId
											+ " prepare succeeded write file cost:"
											+ (System.currentTimeMillis() - ct));
								}
							});

					if (result != -1) {// failed
						PrepareResponseBad.Builder response = PrepareResponseBad
								.newBuilder().setLastPrepareId(result);
						responseBuilder = MessagePackage.newBuilder();
						responseBuilder.setPrepareResponseBad(response);
						logger.debug("prepare refused: instanceId "
								+ request.getInstanceId()
								+ "  conflict with highestJoinNum:" + result
								+ " exceptedJoinNum:"
								+ request.getId().getIncreaseId() + ":"
								+ request.getId().getAddress());
					}
				}
			} catch (Exception e) {
				logger.error("fatal error", e);
			}
		} else if (messagePackage.hasInstanceVoteRequest()) {
			InstanceVoteRequest request = messagePackage
					.getInstanceVoteRequest();
			logger.debug("vote instanceId:" + request.getInstanceId());
			final long ct = System.currentTimeMillis();
			PrepareRecordRealWriteEvent realWriteEvent = new PrepareRecordRealWriteEvent() {
				@Override
				public void call(long instanceId, InstanceVoteRecord record) {
					VoteResponse.Builder response = VoteResponse.newBuilder()
							.setRefuseId(-1);
					MessagePackage.Builder builder = MessagePackage
							.newBuilder();
					builder.setVoteResponse(response);
					// send response
					sendResponse(session, builder, messagePackage.getId());
					logger.debug("isntanceId:" + instanceId
							+ "vote succeed write file cost:"
							+ (System.currentTimeMillis() - ct));
				}
			};
			try {
				long result = aBook
						.writeVotedRecord(request.getInstanceId(),
								request.getVoteId(), request.getValue(),
								realWriteEvent);
				if (result != -1) {
					VoteResponse.Builder response = VoteResponse.newBuilder()
							.setRefuseId(result);
					responseBuilder = MessagePackage.newBuilder();
					responseBuilder.setVoteResponse(response);
					logger.debug("isntanceId:" + request.getInstanceId()
							+ " vote refused");
				} else {
					logger.debug("prepare wait real write event");
				}
			} catch (IOException e) {
				logger.error("fatal error", e);
			}
		} else if (messagePackage.hasGetNewsSenatorsRequest()) {
			NodesCollection senators = cState.getSenators();
			SenatorFollowerVersion version = senatorVersion;
			GetNewsSenatorsResponse.Builder builder = GetNewsSenatorsResponse
					.newBuilder();
			builder.setVersion(version.version).setMaster(
					cState.getLastElectionId().getId());
			for (ConfigNode s : version.senator) {
				builder.addSenators(s.getIp() + ":" + s.getPort());
			}
			//
			// response = new GetNewsSenatorsResponse(version.version,
			// (HashSet<String>) version.senator,
			// cState.getLastElectionId());
			responseBuilder = MessagePackage.newBuilder();
			responseBuilder.setGetNewsSenatorsResponse(builder);
		} else if (messagePackage.hasCatchUpRequest()) {
			CatchUpRequest request = messagePackage.getCatchUpRequest();
			long startId = request.getStartInstanceId();
			int size = request.getSize();
			final CatchUpResponse.Builder b = CatchUpResponse.newBuilder();
			try {
				logger.debug("catchup request startId:" + startId + " size:" + size);
				aBook.readSuccessRecord(startId, startId + size - 1,
						new ReadSuccessRecordCallback() {
							int packageSize = 0;

							@Override
							public void read(
									SuccessfulRecordWrap successfulRecordWrap) {
								CatchUpRecord catchUpRecord = CatchUpRecord
										.newBuilder()
										.setInstanceId(
												successfulRecordWrap
														.getInstanceId())
										.setRecord(
												successfulRecordWrap
														.getRecord()).build();
								int recordSize = catchUpRecord
										.getSerializedSize();
								if (packageSize + recordSize >= MAX_CATCH_SIZE
										&& b.getSuccessfulRecordsCount() != 0) {
									MessagePackage.Builder response = MessagePackage
											.newBuilder().setCatchUpResponse(
													b.build());
									sendResponse(session, response,
											messagePackage.getId(), false);
									response.clear();
									b.clear();
									packageSize = 0;
								}
								packageSize += recordSize;
								b.addSuccessfulRecords(catchUpRecord);

							}
						});
			} catch (IOException e) {
				logger.error("fatal error", e);
			}
			responseBuilder = MessagePackage.newBuilder().setCatchUpResponse(
					b.build());
		}
		sendResponse(session, responseBuilder, messagePackage.getId());
	}

	private void sendResponse(ProtocolHandlerIoSession session,
			MessagePackage.Builder responseBuilder, long responseId,
			boolean isLast) {
		if (responseBuilder == null)
			return;
		responseBuilder.setId(responseId);
		responseBuilder.setIsLast(isLast);
		try {
			session.writeObject(responseBuilder.build());
		} catch (SessionHavaClosedException e) {
		}
	}

	private void sendResponse(ProtocolHandlerIoSession session,
			MessagePackage.Builder responseBuilder, long responseId) {
		sendResponse(session, responseBuilder, responseId, true);
	}

	// private void sendResponse(Serializable response, ProtocolHandlerIoSession
	// session, MessagePackage mp) {
	// if (response != null) {
	// try {
	// MessagePackage responseMp = new MessagePackage(mp.getId(),
	// response, true);
	// session.writeObject(responseMp);
	// } catch (SessionHavaClosedException e) {
	// }
	// }
	// }

	@Override
	public void onOneThreadMessageSent(ProtocolHandlerIoSession attachment,
			int sendSize) {

	}
}
